package com.atguigu.gmall.realtime.common.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.HBaseUtil;
import com.atguigu.gmall.realtime.common.util.RedisUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.hadoop.hbase.client.Connection;
import redis.clients.jedis.Jedis;


/**
 * ClassName: DimFunction
 * Description:
 * 旁路缓存关联维度模板
 *
 * @Create 2024/4/27 9:05
 */
public abstract class DimMapFunction<T> extends RichMapFunction<T, T> implements DimFunction<T> {

    private Connection hBaseConnection;

    private Jedis jedis;

    @Override
    public void open(Configuration parameters) throws Exception {
        hBaseConnection = HBaseUtil.getHBaseConnection();
        jedis = RedisUtil.getJedis();
    }

    @Override
    public void close() throws Exception {
        HBaseUtil.closeHBaseConn(hBaseConnection);
        RedisUtil.closeJedis(jedis);
    }

    @Override
    public T map(T obj) throws Exception {
        //根据流中对象获取要关联的维度的主键
        String key = getRowKey(obj);

        //从Redis中获取要关联的维度
        JSONObject dimJsonObj = RedisUtil.readDim(jedis, getTableName(), key);

        if (dimJsonObj != null) {

            //如果从Redis中获取到了维度数据（缓存命中）
            System.out.println("从Redis中获取" + getTableName() + "的" + key + "数据");

        } else {

            //如果没有获取到，发送请求到HBase中查询数据
            dimJsonObj = HBaseUtil.getRow(hBaseConnection, Constant.HBASE_NAMESPACE, getTableName(), key, JSONObject.class);

            if (dimJsonObj != null) {

                System.out.println("从HBase中获取" + getTableName() + "的" + key + "数据");

                //将查询出来的维度放到Redis中缓存起来
                RedisUtil.writeDim(jedis, getTableName(), key, dimJsonObj);

            } else {

                System.out.println("从Hbase中没有找到" + getTableName() + "的" + key + "数据");

            }
        }

        //将维度对象相关的属性补充到流中对象上
        if (dimJsonObj != null) {

            addDims(obj, dimJsonObj);

        }

        return obj;
    }

    //模板无法实现的方法，交给子类继承实现
    public abstract void addDims(T obj, JSONObject dimJsonObj);

    public abstract String getTableName();

    public abstract String getRowKey(T obj);


}
